package com.its.common.net.server.impl;

import com.its.common.net.core.handler.TailHandler;
import com.its.common.net.server.BaseServer;
import com.its.common.net.server.channel.SocketAddrManager;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioDatagramChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.ReferenceCountUtil;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * @Author xiaxp
 * @Date 9/14/22 15:24
 * @Description
 */
@Slf4j
public class UdpServer extends BaseServer<InetSocketAddress> {

    SocketAddrManager socketAddrManager;

    public UdpServer(int port, List<SimpleChannelInboundHandler> handlers, long duration, TimeUnit unit) {
        super(port, handlers);
        socketAddrManager = SocketAddrManager.getInstance();
        socketAddrManager.init(duration, unit);
    }

    @Override
    public ChannelFuture doStart(int port) throws InterruptedException {
        boss = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(boss)
            .handler(new LoggingHandler(LogLevel.INFO))
            .channel(NioDatagramChannel.class)
            .option(ChannelOption.SO_BROADCAST, true)
            .option(ChannelOption.SO_SNDBUF, READ_WRITE_BUF_CAP)
            .option(ChannelOption.SO_RCVBUF, READ_WRITE_BUF_CAP)
            .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
            .handler(new ChannelInitializer<NioDatagramChannel>() {
                @Override
                protected void initChannel(NioDatagramChannel channel) throws Exception {
                    ChannelPipeline pipeline = channel.pipeline();

                    handlers.forEach((handler) -> {
                        pipeline.addLast(defaultEventExecutorGroup, handler);
                    });

                    pipeline.addLast(defaultEventExecutorGroup, new TailHandler());
                }
            });

        ChannelFuture future = bootstrap.bind(port).sync();
        return future;
    }

    @Override
    public void doAfterStartUp(Channel channel) {
    }


    @Override
    public void sendToAll(ByteBuf content) {
        Collection<InetSocketAddress> clientAddrs = socketAddrManager.getAllInetSocketAddress();
        //TODO ByteBuf释放
        clientAddrs.stream().forEach(address -> send(address, Unpooled.copiedBuffer(content)));
        if (content.refCnt() > 0) {
            ReferenceCountUtil.safeRelease(content);
        }
    }
}